package com.xcn.hadoop.webhdfs;


import java.io.*;
import java.net.*;
import java.text.*;

import net.sf.json.*;

import org.slf4j.*;



/**
 * 
===== HTTP GET <br/>
<li>OPEN (see FileSystem.open)
<li>GETFILESTATUS (see FileSystem.getFileStatus)
<li>LISTSTATUS (see FileSystem.listStatus)
<li>GETCONTENTSUMMARY (see FileSystem.getContentSummary)
<li>GETFILECHECKSUM (see FileSystem.getFileChecksum)
<li>GETHOMEDIRECTORY (see FileSystem.getHomeDirectory)
<li>GETDELEGATIONTOKEN (see FileSystem.getDelegationToken)
<li>GETDELEGATIONTOKENS (see FileSystem.getDelegationTokens)
<br/>
===== HTTP PUT <br/>
<li>CREATE (see FileSystem.create)
<li>MKDIRS (see FileSystem.mkdirs)
<li>CREATESYMLINK (see FileContext.createSymlink)
<li>RENAME (see FileSystem.rename)
<li>SETREPLICATION (see FileSystem.setReplication)
<li>SETOWNER (see FileSystem.setOwner)
<li>SETPERMISSION (see FileSystem.setPermission)
<li>SETTIMES (see FileSystem.setTimes)
<li>RENEWDELEGATIONTOKEN (see FileSystem.renewDelegationToken)
<li>CANCELDELEGATIONTOKEN (see FileSystem.cancelDelegationToken)
<br/>
===== HTTP POST <br/>
APPEND (see FileSystem.append)
<br/>
===== HTTP DELETE <br/>
DELETE (see FileSystem.delete)

 */
public class KerberosWebHDFSConnection implements WebHDFSConnection
{

	private final Logger LOGGER = LoggerFactory.getLogger(KerberosWebHDFSConnection.class);

	private String httpfsUrl = WebHDFSConnectionFactory.DEFAULT_PROTOCOL + WebHDFSConnectionFactory.DEFAULT_HOST + ":" + WebHDFSConnectionFactory.DEFAULT_PORT + WebHDFSConnectionFactory.DEFAULT_CONTEXT_ROOT;
	
	private String userName = WebHDFSConnectionFactory.DEFAULT_USERNAME;
	
	private String token = WebHDFSConnectionFactory.DEFAULT_TOKEN;
	
	public KerberosWebHDFSConnection( )
	{
	}
	
	public KerberosWebHDFSConnection( String httpfsUrl, String userName )
	{
		this.httpfsUrl = httpfsUrl;
		this.userName = userName;
	}
	
	public KerberosWebHDFSConnection( String httpfsUrl, String userName, String token )
	{
		this.httpfsUrl = httpfsUrl;
		this.userName = userName;
		this.token = token;
	}
	
	protected static long copy ( InputStream input, OutputStream result ) throws IOException
	{
		byte[] buffer = new byte[12288]; // 8K=8192 12K=12288 64K=
		long count = 0L;
		int n = 0;
		while ( -1 != (n = input.read ( buffer )) )
		{
			result.write ( buffer, 0, n );
			count += n;
			result.flush ( );
		}
		result.flush ( );
		return count;
	}
	
	/**
	 * Report the result in JSON way
	 * 
	 * @param conn
	 * @param input
	 * @return
	 * @throws IOException
	 */
	private JSONObject result ( HttpURLConnection conn, boolean input ) throws IOException
	{
		StringBuffer sb = new StringBuffer ( );
		if ( input )
		{
			if ( conn.getResponseCode ( ) >= 200 && conn.getResponseCode ( ) <= 300 )
			{
				InputStream is = conn.getInputStream ( );
				BufferedReader reader = new BufferedReader ( new InputStreamReader ( is, "utf-8" ) );
				String line = null;
				while ( (line = reader.readLine ( )) != null )
				{
					sb.append ( line );
				}
				reader.close ( );
				is.close ( );
			}
		}
		JSONObject result = new JSONObject ( );
		result.put ( "code", conn.getResponseCode ( ) );
		result.put ( "mesg", conn.getResponseMessage ( ) );
		result.put ( "type", conn.getContentType ( ) );
		result.put ( "data", sb.toString ( ) );
		
		LOGGER.debug ( result.toString ( ) );

		return result;
	}
	
	
	public String token ( )
	{
		return URLUtil.encodePath ( "&user.name=" + userName + token );
	}
	
	
	/*
	 * ========================================================================
	 * GET
	 * ========================================================================
	 */
	/**
	 * <b>GETHOMEDIRECTORY</b>
	 * 
	 * curl -i "http://<HOST>:<PORT>/webhdfs/v1/?op=GETHOMEDIRECTORY"
	 * 
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 */
	public JSONObject getHomeDirectory ( ) throws MalformedURLException, IOException
	{
		URL url = new URL ( new URL ( httpfsUrl ), "?op=GETHOMEDIRECTORY" + token ( ) );

		LOGGER.info ( "getHomeDirectory : " + url );

		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	

	/**
	 * <b>OPEN</b>
	 * 
	 * curl -i -L "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=OPEN
	 * [&offset=<LONG>][&length=<LONG>][&buffersize=<INT>]"
	 * 
	 * @param path
	 * @param os
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public JSONObject open ( String path, OutputStream os ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=OPEN" + token ( ), URLUtil.encodePath ( path ) ) );
		
		LOGGER.info ( "open : " + url );

		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "GET" );
		conn.setRequestProperty ( "Accept", "text/html, application/xhtml+xml, */*" );
		conn.setRequestProperty ( "User-Agent", "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; WOW64; Trident/5.0)" );
		conn.setRequestProperty ( "Host", "search.naver.com" );
		conn.setRequestProperty ( "Content-Type", "application/octet-stream" );
		conn.connect ( );

		if ( conn.getResponseCode ( ) >= 200 && conn.getResponseCode ( ) <= 300 )
		{
			InputStream is = conn.getInputStream ( );
			copy ( is, os );
			is.close ( );
		}
		os.close ( );

		JSONObject resp = result ( conn, false );
		conn.disconnect ( );
		return resp;
	}
	
	/**
	 * <b>GETCONTENTSUMMARY</b>
	 * 
	 * curl -i "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=GETCONTENTSUMMARY"
	 * 
	 * @param path
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 */
	public JSONObject getContentSummary ( String path ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=GETCONTENTSUMMARY" + token ( ), URLUtil.encodePath ( path ) ) );

		LOGGER.info ( "getContentSummary : " + url );

		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "GET" );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	/**
	 * <b>LISTSTATUS</b>
	 * 
	 * curl -i "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=LISTSTATUS"
	 * 
	 * @param path
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 */
	public JSONObject listStatus ( String path ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=LISTSTATUS" + token ( ), URLUtil.encodePath ( path ) ) );
		
		LOGGER.info ( "listStatus : " + url );
		
		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "GET" );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	
	/**
	 * <b>LISTSTATUS</b>
	 * 
	 * curl -i "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=LISTSTATUS"
	 * 
	 * @param path
	 * @param detail
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 */
	public JSONArray listStatus( String path, boolean detail ) throws MalformedURLException, IOException
	{
		JSONObject resp = listStatus ( path );
		JSONObject obj1 = ( JSONObject ) JSONSerializer.toJSON ( resp.getString ( "data" ) );
		JSONObject obj2 = ( JSONObject ) JSONSerializer.toJSON ( obj1.getString ( "FileStatuses" ) );
		JSONArray fileList = ( JSONArray ) JSONSerializer.toJSON ( obj2.get ( "FileStatus" ) );
		return fileList;
	}
	
	
	
	/**
	 * <b>GETFILESTATUS</b>
	 * 
	 * curl -i "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=GETFILESTATUS"
	 * 
	 * @param path
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 */
	public JSONObject getFileStatus ( String path ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=GETFILESTATUS" + token ( ), URLUtil.encodePath ( path ) ) );
		
		LOGGER.info ( "getFileStatus : " + url );
		
		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "GET" );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	/**
	 * <b>GETFILECHECKSUM</b>
	 * 
	 * curl -i "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=GETFILECHECKSUM"
	 * 
	 * @param path
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 */
	public JSONObject getFileCheckSum ( String path ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=GETFILECHECKSUM" + token ( ), URLUtil.encodePath ( path ) ) );
		
		LOGGER.info ( "getFileCheckSum : " + url );
		
		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "GET" );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	/*
	 * ========================================================================
	 * PUT
	 * ========================================================================
	 */
	/**
	 * <b>CREATE</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=CREATE
	 * [&overwrite=<true|false>][&blocksize=<LONG>][&replication=<SHORT>]
	 * [&permission=<OCTAL>][&buffersize=<INT>]"
	 * 
	 * @param path
	 * @param is
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 */
	public JSONObject create ( String path, InputStream is ) throws MalformedURLException, IOException
	{
		JSONObject resp = null;
		
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=CREATE" + token ( ), URLUtil.encodePath ( path ) ) );

		LOGGER.info ( "create : " + url );

		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "PUT" );
		conn.setInstanceFollowRedirects ( false );
		conn.connect ( );
		resp = result ( conn, true );
		
		String redirectUrl = null;
		if ( conn.getResponseCode ( ) == 307 ) redirectUrl = conn.getHeaderField ( "Location" );
		conn.disconnect ( );

		if ( redirectUrl != null )
		{
			conn = ( HttpURLConnection ) new URL ( redirectUrl ).openConnection ( );
			conn.setRequestMethod ( "PUT" );
			conn.setDoOutput ( true );
			conn.setDoInput ( true );
			conn.setUseCaches ( false );
			conn.setRequestProperty ( "Content-Type", "application/octet-stream" );
			final int _SIZE = is.available ( );
			conn.setRequestProperty ( "Content-Length", "" + _SIZE );
			conn.setFixedLengthStreamingMode ( _SIZE );
			conn.connect ( );
			OutputStream os = conn.getOutputStream ( );
			copy ( is, os );
			is.close ( );
			os.close ( );
			resp = result ( conn, false );
			conn.disconnect ( );
		}
		
		return resp;
	}
	
	/**
	 * <b>MKDIRS</b>
	 * 
	 * curl -i -X PUT
	 * "http://<HOST>:<PORT>/<PATH>?op=MKDIRS[&permission=<OCTAL>]"
	 * 
	 * @param path
	 * @return
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public JSONObject mkdirs ( String path ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=MKDIRS" + token ( ), URLUtil.encodePath ( path ) ) );
		
		LOGGER.info ( "mkdirs : " + url );
		
		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "PUT" );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	/**
	 * <b>CREATESYMLINK</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/<PATH>?op=CREATESYMLINK
	 * &destination=<PATH>[&createParent=<true|false>]"
	 * 
	 * @param path
	 * @return
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public JSONObject createSymLink ( String srcPath, String destPath ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=CREATESYMLINK&destination={1}" + token ( ), URLUtil.encodePath ( srcPath ), URLUtil.encodePath ( destPath ) ) );
		
		LOGGER.info ( "createSymLink : " + url );
		
		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "PUT" );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	/**
	 * <b>RENAME</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/<PATH>?op=RENAME
	 * &destination=<PATH>[&createParent=<true|false>]"
	 * 
	 * @param path
	 * @return
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public JSONObject rename ( String srcPath, String destPath ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=RENAME&destination={1}" + token ( ), URLUtil.encodePath ( srcPath ), URLUtil.encodePath ( destPath ) ) );
		
		LOGGER.info ( "rename : " + url );
		
		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "PUT" );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	/**
	 * <b>SETPERMISSION</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=SETPERMISSION
	 * [&permission=<OCTAL>]"
	 * 
	 * @param path
	 * @return
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public JSONObject setPermission ( String path ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=SETPERMISSION" + token ( ), URLUtil.encodePath ( path ) ) );
		
		LOGGER.info ( "setPermission : " + url );
		
		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "PUT" );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	/**
	 * <b>SETOWNER</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=SETOWNER
	 * [&owner=<USER>][&group=<GROUP>]"
	 * 
	 * @param path
	 * @return
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public JSONObject setOwner ( String path ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=SETOWNER" + token ( ), URLUtil.encodePath ( path ) ) );
		
		LOGGER.info ( "setOwner : " + url );

		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "PUT" );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	/**
	 * <b>SETREPLICATION</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=SETREPLICATION
	 * [&replication=<SHORT>]"
	 * 
	 * @param path
	 * @return
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public JSONObject setReplication ( String path ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=SETREPLICATION" + token ( ), URLUtil.encodePath ( path ) ) );
		
		LOGGER.info ( "setReplication : " + url );

		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "PUT" );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	/**
	 * <b>SETTIMES</b>
	 * 
	 * curl -i -X PUT "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=SETTIMES
	 * [&modificationtime=<TIME>][&accesstime=<TIME>]"
	 * 
	 * @param path
	 * @return
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public JSONObject setTimes ( String path ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=SETTIMES" + token ( ), URLUtil.encodePath ( path ) ) );
		
		LOGGER.info ( "setTimes : " + url );

		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "PUT" );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	/*
	 * ========================================================================
	 * POST
	 * ========================================================================
	 */
	/**
	 * curl -i -X POST
	 * "http://<HOST>:<PORT>/webhdfs/v1/<PATH>?op=APPEND[&buffersize=<INT>]"
	 * 
	 * @param path
	 * @param is
	 * @return
	 * @throws MalformedURLException
	 * @throws IOException
	 
	 */
	public JSONObject append ( String path, InputStream is ) throws MalformedURLException, IOException
	{
		JSONObject resp = null;

		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=APPEND" + token ( ), URLUtil.encodePath ( path ) ) );
		
		LOGGER.info ( "append : " + url );

		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "POST" );
		conn.setInstanceFollowRedirects ( false );
		conn.connect ( );
		
		LOGGER.info ( "Location : " + conn.getHeaderField ( "Location" ) );
		
		String redirectUrl = null;
		resp = result ( conn, true );
		if ( conn.getResponseCode ( ) == 307 ) redirectUrl = conn.getHeaderField ( "Location" );
		conn.disconnect ( );

		if ( redirectUrl != null )
		{
			conn = ( HttpURLConnection ) new URL ( redirectUrl + token ( ) ).openConnection ( );
			conn.setRequestMethod ( "POST" );
			conn.setDoOutput ( true );
			conn.setDoInput ( true );
			conn.setUseCaches ( false );
			conn.setRequestProperty ( "Content-Type", "application/octet-stream" );
			final int _SIZE = is.available ( );
			conn.setRequestProperty ( "Content-Length", "" + _SIZE );
			conn.setFixedLengthStreamingMode ( _SIZE );
			conn.connect ( );
			OutputStream os = conn.getOutputStream ( );
			copy ( is, os );
			is.close ( );
			os.close ( );
			resp = result ( conn, true );
			conn.disconnect ( );
		}
		return resp;
	}
	
	/*
	 * ========================================================================
	 * DELETE
	 * ========================================================================
	 */
	/**
	 * <b>DELETE</b>
	 * 
	 * curl -i -X DELETE "http://<host>:<port>/webhdfs/v1/<path>?op=DELETE
	 * [&recursive=<true|false>]"
	 * 
	 * @param path
	 * @return
	 * @throws IOException
	 * @throws MalformedURLException
	 */
	public JSONObject delete ( String path ) throws MalformedURLException, IOException
	{
		URL url = new URL ( httpfsUrl + MessageFormat.format ( "{0}?op=DELETE" + token ( ), URLUtil.encodePath ( path ) ) );
		
		LOGGER.info ( "delete : " + url );
		
		HttpURLConnection conn = ( HttpURLConnection ) url.openConnection ( );
		conn.setRequestMethod ( "DELETE" );
		conn.setInstanceFollowRedirects ( false );
		conn.connect ( );

		JSONObject resp = result ( conn, true );
		conn.disconnect ( );
		return resp;
	}
	
	// Begin Getter & Setter
	public String getHttpfsUrl ( )
	{
		return httpfsUrl;
	}
	
	public void setHttpfsUrl ( String httpfsUrl )
	{
		this.httpfsUrl = httpfsUrl;
	}

	public String getUserName ( )
	{
		return userName;
	}

	public void setUserName ( String userName )
	{
		this.userName = userName;
	}
	// End Getter & Setter
	
	
	public static void main ( String[] args ) throws Exception
	{
		if ( args.length == 0 )
		{
			System.err.println ( "Web HDFS args..\n1. url\n2. user name\n3. command(OPEN, DOWN, GETFILECHECKSUM, LISTSTATUS)\n4. path" );
			System.exit ( 0 );
		}

		KerberosWebHDFSConnection _con = new KerberosWebHDFSConnection ( args[0], args[1] );
		if ( args[2].equals ( "OPEN" ) )
		{
			ByteArrayOutputStream os = new ByteArrayOutputStream ( );
			System.out.println ( _con.open ( args[3], os ) );
			System.out.println ( os.toString ( ) );
		}
		else if ( args[2].equals ( "DOWN" ) )
		{
			String fileName = args[3].substring ( args[3].lastIndexOf ( "/" ) + 1 );
			File _file = new File ( fileName );
			FileOutputStream fos = new FileOutputStream ( _file );
			System.out.println ( _con.open ( args[3], fos ) );
			System.out.println ( _file.getAbsolutePath ( ) );
		}
		else if ( args[2].equals ( "MKDIRS" ) )
			System.out.println ( _con.mkdirs ( args[3] ) );
		else if ( args[2].equals ( "GETFILECHECKSUM" ) )
			System.out.println ( _con.getFileCheckSum ( args[3] ) );
		else if ( args[2].equals ( "LISTSTATUS" ) )
			System.out.println ( _con.listStatus ( args[3], true ) );
		
		System.out.println ( "" );
	}
}
